package com.zjw.shortlink.project.config;

import com.zjw.shortlink.project.mq.consumer.ShortLinkStatsSaveConsumer;
import lombok.RequiredArgsConstructor;
import org.apache.ibatis.javassist.bytecode.analysis.Executor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

//@Configuration
@RequiredArgsConstructor
public class RedisStreamConfiguration {
    private final RedisConnectionFactory redisConnectionFactory;
    private final ShortLinkStatsSaveConsumer shortLinkStatsSaveConsumer;

    @Value("${spring.data.redis.channel-topic.short-link-stats-group}")
    private String group;

    @Value("${spring.data.redis.channel-topic.short-link-stats}")
    private String topic;
    @Bean
    public ExecutorService asyncStreamConsumer() {
        AtomicInteger index = new AtomicInteger(0);
        int availableProcessors = Runtime.getRuntime().availableProcessors();
        return new ThreadPoolExecutor(
                availableProcessors,
                availableProcessors + availableProcessors >> 1,
                3,
                TimeUnit.MINUTES,
                new LinkedBlockingQueue<>(),
                (Runnable r) -> {
                    Thread thread = new Thread(r);
                    thread.setName("stream_consumer_short-link_stats_" + index.incrementAndGet());
                    thread.setDaemon(true);
                    return thread;
                }
        );
    }
    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(ExecutorService asyncStreamConsumer) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                .batchSize(10)
                .executor(asyncStreamConsumer)
                .pollTimeout(Duration.ofMillis(500))
                .build();
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> messageListenerContainer =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);
//        messageListenerContainer.receiveAutoAck(
//                Consumer.from(group, "stats-consumer"),
//                StreamOffset.create(topic, ReadOffset.lastConsumed()),
//                shortLinkStatsSaveConsumer
//        );
        return messageListenerContainer;
    }
}
